package com.atguigu.flink.watermark;

import com.atguigu.flink.function.WaterSensorMapFunction;
import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.TimerService;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;

/**
 * Created by Smexy on 2023/4/8
 *
 *  你今年明明已经60岁了，但是看起来还跟18岁的小姑娘一样。
 *  你今年明明已经61岁了，但是看起来还跟18岁的小姑娘一样。
 *  你今年明明已经62岁了，但是看起来还跟82岁的小姑娘一样。
 *      问题： 出生日期是什么时候?
 *              主观： 18年前出生
 *              客观:  60岁前出生
 *
 *  重生之我在明朝搭数仓，今年23岁，穿越到了1700年了，你到底多大?
 *              主观： 23年岁
 *              客观:  -300岁
 *
 *
 *
 * ----------------------------------
 *  主观时间： 人认为的时间，主观性很强。
 *                 可以停滞
 *                 速度不恒定
 *
 *
 *  客观时间:  客观世界运行的时间。
 *                  不以人的意志为转移。
 *                  永不停滞
 *                  速度恒定
 *
 *
 * -------------------------------------------
 *  时间语义： 如何去获取时间。
 *      flink:
 *          processingTime: 参考计算机的物理时钟获取时间。
 *                      客观时间。
 *
 *          eventTime:     参考数据中提供的时间属性获取时间。
 *                     主观时间。
 *
 *   eventtime最常用。
 *          离线数仓： 当天把昨天生成的数据，导入到数仓。放在 昨天的 分区。
 *                   6-15当天凌晨，运行采集程序，把6-14日生成的数据，导入到数仓，放在每一层的 6-14的分区。
 *  ----------------------------------------------
 *    watermark: 水位线，水印。
 *    在flink中，如果是eventtime语义，此时flink使用watermark作为时间。
 *
 *    数据(ts = 100)----->提取时间属性ts=100------>再加工成flink中的watermark。
 *    把watermark放入流中，向下游传递，传递以后，下游的算子，就知道当前几点了。
 *
 *    -------------------
 *      watermark特征:
 *          public Watermark(long timestamp)
 *          ①一个watermark中包含了一个时间属性
 *          ②可以从源头产生，Source算子读数据时，产生
 *            可以从流的某个中间位置产生
 *          ③在产生以后，向下游流动，通知下游的算子，当前系统时间是几点了
 *          ④WaterMark作为一种特殊的数据，随着你的数据流，向下游流动。
 *              WaterMark总是落后于数据一个身位。
 *
 *      有什么用?
 *          基于eventtime的语义下，在某些需要时间的运算中，才有用。
 *          ①时间窗口的运算
 *          ②定时器
 *
 *
 *---------------------------------------------
 *  assignTimestampsAndWatermarks(WatermarkStrategy x): 帮你产生水印。
 *
 *        WatermarkStrategy：  水印策略
 *              ①如何从数据中提取时间属性
 *
 *              ②如何对时间属性再进行加工
 *                      forMonotonousTimestamps()    ： 时间属性 - 1ms
 *                      forBoundedOutOfOrderness(Duration maxOutOfOrderness):  时间属性 - 1ms -  maxOutOfOrderness
 *
 *
 * -----------------------------------------
 *  刚上手，一定要把并行度设置为1。
 *
 *  --------------------------
 *      Watermark就是时间，时间不可回退，单调递增。
 *          时间不能逆流。
 *
 *          时间是一条逆流的河。
 *              河流:  从高处---->低处
 *                      单调递减
 *
 *              时间:  从小---->大
 *                    过去----->未来
 *                      单调递增
 *
 *       watermark只会取当前收到的所有的事件时间属性中的最大值，来制作水印。
 *          数据(ts=100) ---> watermark(100-1)
 *          数据(ts=200) ---> watermark(200-1)
 *          数据(ts=50) ---> watermark(200-1)
 *
 *
 *
 *
 */
public class Demo1_WaterMarkView
{
    public static void main(String[] args) {
        
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 3333);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);

        //设置周期性发送水印的时间间隔
        env.getConfig().setAutoWatermarkInterval(2000);

        env.disableOperatorChaining();

        env.setParallelism(1);

        //从一个 WaterSensor element 提取 时间戳
        WatermarkStrategy<WaterSensor> watermarkStrategy = WatermarkStrategy
            //提取了时间戳后，怎么加工
            .<WaterSensor>forMonotonousTimestamps()
            //怎么去提取数据中的时间属性  返回的必须是毫秒
            .withTimestampAssigner( (e, ts) -> e.getTs());


        env
                   .socketTextStream("hadoop102", 8888)
                   .map(new WaterSensorMapFunction())
                   //产生了水印，向下游发送水印
                   .assignTimestampsAndWatermarks(watermarkStrategy)
                   .process(new ProcessFunction<WaterSensor, WaterSensor>()
                   {
                       /*
                        来一条数据，就调用一次
                        */
                       @Override
                       public void processElement(WaterSensor value, Context ctx, Collector<WaterSensor> out) throws Exception {

                           //当前算子的时钟服务
                           TimerService timerService = ctx.timerService();
                           //当前数据到达时，此时process算子的watermark(时间)
                           System.out.println("主观时间(eventtime):"+timerService.currentWatermark());
                           System.out.println("客观时间(processingtime):"+timerService.currentProcessingTime());
                           out.collect(value);

                       }
                   })
                   .print();

        
                try {
                            env.execute();
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
        
    }
}
